package dy.com.第7章服务器优化.Server;


import dy.com.收发并行.Util.CloseUtils;
import dy.com.第7章服务器优化.Server.handler.ClientHandler;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TCPServer implements ClientHandler.ClientHandlerCallback {
    private ClientListener clientListener = null;
    int listenPort;
    private List<ClientHandler> clientHandlers = new ArrayList<>();
    private final ExecutorService forwardingThreadPoolExcutor;
    Selector selector;
    ServerSocketChannel serverSocketChannel;

    public TCPServer(int listenPort) {
        this.listenPort = listenPort;
        this.forwardingThreadPoolExcutor = Executors.newSingleThreadExecutor();
    }

    void start() throws IOException {
        //先停止监听
        //stop();

        //新建selector
        selector = Selector.open();
        //创建用于监听的ServerSocketChannel
        serverSocketChannel = ServerSocketChannel.open();
        //设置为非阻塞
        serverSocketChannel.configureBlocking(false);
        //channel绑定端口
        serverSocketChannel.socket().bind(new InetSocketAddress(listenPort));
        //注册channel到到selector
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);


        System.out.println("tcp服务器信息:ip:" + serverSocketChannel.getLocalAddress().toString());


        System.out.println("启动监听线程");
        clientListener = new ClientListener();
        clientListener.start();

    }

    //停止监听线程
    void stop() {
        if (clientListener != null) {
            clientListener.exit();
            clientListener = null;
        }
        CloseUtils.close(serverSocketChannel);
        CloseUtils.close(selector);

        synchronized (TCPServer.this) {
            for (ClientHandler clientHandler : clientHandlers) {
                clientHandler.exit();
            }

            clientHandlers.clear();
        }
        //关闭线程池
        forwardingThreadPoolExcutor.shutdown();
    }

    public synchronized void broadcast(String str) {
        for (ClientHandler clientHandler : clientHandlers) {
            clientHandler.send(str);
        }
    }

    @Override
    public synchronized void onSelfClosed(ClientHandler clientHandler) {
        clientHandlers.remove(clientHandler);
    }

    @Override
    public void onNewMessageArrived(final ClientHandler clientHandler, final String msg) {
     //   System.out.println("Received-" + clientHandler.getClientInfo() + ":" + msg);
        //异步线程池
        forwardingThreadPoolExcutor.execute(() -> {
            synchronized (TCPServer.this) {
                for (ClientHandler clientHandler1 : clientHandlers) {
                    if (clientHandler1.equals(clientHandler)) {
                        continue;
                    }
                    clientHandler1.send(clientHandler.getClientInfo() + ":" + msg);
                }
            }
        });
    }

    //监听线程
    private class ClientListener extends Thread {
        int listenPort;
        boolean finish;

        @Override
        public void run() {
            System.out.println("tcp服务器监听线程正在run");
            while (!finish) {

                try {
                    //select()不设置参数意思是永远阻塞
                    if (selector.select() == 0) {
                        if (finish) {
                            break;
                        }
                        continue;
                    }
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        if (finish) {
                            break;
                        }
                        SelectionKey key = iterator.next();
                        //将遍历到的删除
                        iterator.remove();
                        //检查key是否是到达状态
                        if (key.isAcceptable()) {
                            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                            //得到客户端channel
                            SocketChannel socketChannel = serverSocketChannel.accept();

                            try {
                                ClientHandler clientHandler = new ClientHandler(socketChannel, TCPServer.this);

                                   //添加同步
                                synchronized (TCPServer.this) {
                                    clientHandlers.add(clientHandler);
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                    }

                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("监听服务已经关闭");
        }

        private void exit() {
            finish = true;
            //唤醒阻塞
            selector.wakeup();
        }
    }


}
